package com.king.freefish.online.repository;

import cn.hutool.core.util.ObjectUtil;
import com.king.freefish.entity.OnlineUser;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Repository;

import java.time.LocalDateTime;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @packageName com.king.freefish.repository
 * @author: jinguoguang
 * @date: 2025/7/29 15:33
 * @description:
 **/
@Repository
@AllArgsConstructor
@Slf4j
public class OnlineUserRepository {
    private static final String ONLINE_USERS_KEY = "online_users";
    private static final String ONLINE_IPS_KEY = "online_ips";
    private static final String IP_SESSIONS_KEY_PREFIX = "ip_sessions:";

    // 会话超时时间(秒)
    private static final long SESSION_TIMEOUT = 30;


    private final RedisTemplate<String, Object> redisTemplate;

    /**
     * 用户上线
     */
    public void online(OnlineUser user) {
        String sessionId = user.getSessionId();
        String ip = user.getIp();

        // 使用Redis的Set存储在线IP集合，自动去重
        redisTemplate.opsForSet().add(ONLINE_IPS_KEY, ip);

        // 使用Redis的Hash存储IP对应的所有session
        String ipSessionsKey = IP_SESSIONS_KEY_PREFIX + ip;
        redisTemplate.opsForHash().put(ipSessionsKey, sessionId, LocalDateTime.now().toString());

        // 设置过期时间，防止内存泄漏
        redisTemplate.expire(ipSessionsKey, SESSION_TIMEOUT * 2, TimeUnit.SECONDS);

        // 使用Redis的Hash存储完整的用户会话信息
        redisTemplate.opsForHash().put(ONLINE_USERS_KEY, sessionId, user);
        redisTemplate.expire(ONLINE_USERS_KEY, SESSION_TIMEOUT * 2, TimeUnit.SECONDS);
    }

    /**
     * 用户下线
     */
    public void offline(String sessionId) {
        // 获取用户会话信息
        OnlineUser user = (OnlineUser) redisTemplate.opsForHash().get(ONLINE_USERS_KEY, sessionId);
        if (ObjectUtil.isNull(user)) {
            return;
        }
        String ip = user.getIp();
        String ipSessionsKey = IP_SESSIONS_KEY_PREFIX + ip;
        // 移除IP对应的session
        redisTemplate.opsForHash().delete(ipSessionsKey, sessionId);
        // 检查该IP是否还有其他活跃session
        Long sessionCount = redisTemplate.opsForHash().size(ipSessionsKey);
        if (sessionCount == 0) {
            // 该IP没有活跃session了，从在线IP集合中移除
            redisTemplate.opsForSet().remove(ONLINE_IPS_KEY, ip);
            // 移除IP对应的session记录
            redisTemplate.delete(ipSessionsKey);
        }

        // 移除完整的用户会话信息
        redisTemplate.opsForHash().delete(ONLINE_USERS_KEY, sessionId);
    }

    /**
     * 获取在线IP数量
     */
    public long getOnlineIpCount() {
        try {
            Long count = redisTemplate.opsForSet().size(ONLINE_IPS_KEY);
            return count != null ? count : 0L;
        } catch (Exception e) {
            // 记录异常日志，便于问题追踪
            log.error("获取在线IP数量失败", e);
            return 0L;
        }
    }


    /**
     * 获取在线用户会话数量
     */
    public long getOnlineSessionCount() {
        return redisTemplate.opsForHash().size(ONLINE_USERS_KEY);
    }

    /**
     * 获取所有在线IP
     */
    public Set<Object> getAllOnlineIps() {
        return redisTemplate.opsForSet().members(ONLINE_IPS_KEY);
    }

    /**
     * 获取指定IP的所有会话ID
     */
    public Set<Object> getSessionsByIp(String ip) {
        String ipSessionsKey = IP_SESSIONS_KEY_PREFIX + ip;
        return redisTemplate.opsForHash().keys(ipSessionsKey);
    }

    /**
     * 清理过期会话
     */
    public void cleanExpiredSessions() {
        // 获取所有在线IP
        Set<Object> onlineIps = getAllOnlineIps();
        if (onlineIps == null || onlineIps.isEmpty()) {
            return;
        }
        LocalDateTime now = LocalDateTime.now();
        int totalExpired = 0;
        int totalProcessed = 0;

        // 遍历每个IP的会话，移除过期的会话
        for (Object ip : onlineIps) {
            String ipSessionsKey = IP_SESSIONS_KEY_PREFIX + ip;
            try {
                // 使用Redis Pipeline批量获取所有IP的会话信息，减少网络开销
                List<Object> results = redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                    HashOperations<String, Object, Object> hashOps = redisTemplate.opsForHash();
                    hashOps.entries(ipSessionsKey);
                    return null;
                });

                if (!results.isEmpty()) {
                    @SuppressWarnings("unchecked")
                    Map<Object, Object> sessions = (Map<Object, Object>) results.get(0);
                    totalProcessed += sessions.size();
                    // 并行处理过期会话检测（在多核环境下可能提升性能）
                    List<String> expiredSessionIds = sessions.entrySet().parallelStream()
                            .filter(entry -> entry.getKey() != null && entry.getValue() != null)
                            .filter(entry -> {
                                try {
                                    LocalDateTime loginTime = LocalDateTime.parse((String) entry.getValue());
                                    return loginTime.plusSeconds(SESSION_TIMEOUT).isBefore(now);
                                } catch (Exception e) {
                                    log.warn("Invalid session time format for session: {}", entry.getKey(), e);
                                    return true; // 格式错误的数据也视为过期
                                }
                            })
                            // 提取过期会话ID
                            .map(entry -> (String) entry.getKey())
                            .collect(Collectors.toList());
                    // 累计过期会话数量
                    totalExpired += expiredSessionIds.size();
                    // 使用Redis Pipeline批量删除过期会话
                    if (!expiredSessionIds.isEmpty()) {
                        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                            HashOperations<String, Object, Object> hashOps = redisTemplate.opsForHash();
                            expiredSessionIds.forEach(sessionId -> {
                                try {
                                    hashOps.delete(ipSessionsKey, sessionId);
                                    offline(sessionId);
                                } catch (Exception e) {
                                    log.error("Failed to offline session: {}", sessionId, e);
                                }
                            });
                            return null;
                        });
                    }
                }
            } catch (Exception e) {
                log.error("Failed to process sessions for IP: {}", ip, e);
            }
        }

        // 添加清理统计日志
        log.info("Session cleanup completed. Total processed: {}, Total expired: {}",
                totalProcessed, totalExpired);
    }

}